package HandshakeJoin;

import java.util.List;
import java.util.ArrayList;
import java.util.concurrent.TimeUnit;
import com.google.common.collect.Lists;
import org.apache.s4.core.ProcessingElement;
import org.apache.s4.core.Stream;

public class ResultCollector extends ProcessingElement
{
    private Stream<Snapshot> downStream;
    private List<ArrayList<Integer>> currResults;
    private int thrhNumResults = 256;
    private int thrhEmitOnTime = (int) (thrhNumResults * 0.75);
    private int delayCount = 0;
    private int thrhDelayed = 5;

    @Override
    protected void onCreate() {
        currResults = Lists.newLinkedList();
    }

    public void onEvent(JoinResult result) {
        currResults.add(result.getRecord());

        if (currResults.size() >= thrhNumResults) {
            emitSnapshot();
        }
    }

    public void onTime() {
        if (currResults.size() >= thrhEmitOnTime || delayCount >= thrhDelayed) {
            emitSnapshot();
        }
        else {
            ++delayCount;
        }
    }

    @Override
    protected void onRemove() {}

    private void emitSnapshot() {
        if (currResults.isEmpty()) {
            // nothing to emit
            return;
        }

        /* Create new space for the upcoming records. */
        List<ArrayList<Integer>> commitResults = currResults;
        currResults = Lists.newLinkedList();

        /* Emit the records. */
        Snapshot snapshot = new Snapshot(getId(), commitResults);
        downStream.put(snapshot);

        /* Clear the accumulated emitting count. */
        delayCount = 0;
    }

    public ResultCollector setDownStream(final Stream<Snapshot> stream) {
        this.downStream = stream;

        return this;
    }

    public ResultCollector setThreshold(final int thrhNumResults) {
        this.thrhNumResults = thrhNumResults;
        thrhEmitOnTime = (int) (thrhNumResults * 0.6);

        return this;
    }

    public ResultCollector setSnapShotInterval(final long interval) {
        setTimerInterval(interval, TimeUnit.MILLISECONDS);

        return this;
    }

    public ResultCollector setSnapshotDelayCount(final int delayCount) {
        this.thrhDelayed = delayCount;

        return this;
    }
}